package com.boot.consumer;

import cn.hutool.core.thread.ThreadFactoryBuilder;
import lombok.SneakyThrows;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * 
 * @author threestrong
 * @since 2022-08-09 12:00:00
 *
 */
@SuppressWarnings("deprecation")
public class RocketConsumerTopicThread extends Thread{

	private Logger logger = LoggerFactory.getLogger(getClass());

	private DefaultMQPushConsumer defaultMQPushConsumer;

	private String topic;

	public RocketConsumerTopicThread(DefaultMQPushConsumer defaultMQPushConsumer, String topic) {
		this.defaultMQPushConsumer = defaultMQPushConsumer;
		this.topic = topic;
	}

	@SneakyThrows
	@SuppressWarnings({"unchecked", "rawtypes" })
	@Override
	public void run() {
		logger.info("topic为:{}的队列线程开启", topic);
		ExecutorService executorService = new ThreadPoolExecutor(10, 20, 60, TimeUnit.SECONDS, new SynchronousQueue<>(), new ThreadPoolExecutor.DiscardPolicy());
		while (true) {
			List<Future> list = new ArrayList();    // 保存本轮的执行结果
			defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
				@Override
				public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messageExts, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
					// 4. 处理消息消息
					for (MessageExt messageExt : messageExts) {
						// 开启线程池处理每一笔请求
						BusinessRouter runnable = new BusinessRouter(defaultMQPushConsumer, messageExt, topic);
						list.add(executorService.submit(runnable));
					}
					// 返回消费状态
					return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
				}
			});
			defaultMQPushConsumer.start();
		}
	 }

	/**
	 * 
	 * @param errorNum 已经存在的错误次数
	 * @param result 当前记录的结果数据
	 * @param errCode 当前队列对应的错误码
	 * @return 当前记录执行完成后的错误次数
	 */
	public int calErrorNum(int errorNum, Map result, String errCode) {
		String code = (String) result.get("STATUS");
		if (!code.equals("") && code != null && code.startsWith(errCode)) {
			errorNum++;
		}
		return errorNum;
	}

	/**
	 * 
	 * @param errorNum 错误次数
	 * @param total 当前批次总的次数
	 * @param sleepTime 睡眠时间
	 * @param limitRate 错误限制速率
	 * @return
	 */
	public void judgeErrorRate(int errorNum, int total, long sleepTime, BigDecimal limitRate) {
		if(total > 0) {
			BigDecimal rate = new BigDecimal(errorNum / total);
			if(rate.compareTo(limitRate) > 0) {
				try {
					Thread.sleep(sleepTime);
				} catch (InterruptedException exception) {
					exception.printStackTrace();
				}
			}
		}
	}
}
